package cn.yj.common.lock.redis;

import cn.yj.common.AppExecutor;
import cn.yj.common.JedisUtils;
import cn.yj.common.lock.IVariableLock;
import cn.yj.tools.exception.ServiceException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.params.SetParams;

import java.io.IOException;
import java.util.Arrays;
import java.util.Timer;
import java.util.concurrent.*;

/**
 * 引入 redisson 去实现吧
 *
 * <dependency>
 * <groupId>org.redisson</groupId>
 * <artifactId>redisson</artifactId>
 * <version>3.8.2</version>
 * </dependency>
 *
 * @author 永健
 * @since 2022-01-25 16:55
 */
public class RedisVariableLock implements IVariableLock<String> {
    private static final Logger log = LoggerFactory.getLogger(KeyWatchdogTask.class);

    /**
     * 最大持有锁时间
     */
    private static final int KEY_EXPIRE_TIME_SECONDS = 30;

    /**
     * 获取锁的最大超时时间
     */
    private static final int TRY_LOCK_TIME_SECONDS = 60;

    private static final int DELAY_WATCH_DOG = KEY_EXPIRE_TIME_SECONDS / 3;

    private static final String LOCK_RESULT_SUCCESS = "OK";

    private static final Integer UNLOCK_RESULT_SUCCESS = 1;

    private static final Integer UNLOCK_RESULT_ERROR = 0;

    private static final SetParams params = SetParams.setParams().nx().ex(KEY_EXPIRE_TIME_SECONDS);

    @Override
    public void lock(String val) {
        // log.info(Thread.currentThread().getName() + "：尝试取锁.....");
        if (!tryLock(val)) {
            tryLockWait(val);
            return;
        }
    }

    private boolean tryLock(String val) {
        Jedis jedis = JedisUtils.getJedis();
        boolean result = LOCK_RESULT_SUCCESS.equals(jedis.set(val, getKeyValue(val), params));
        jedis.close();
        if (result) {
            log.info(Thread.currentThread().getName() + "：获取锁成功.....");
            // 开启看门狗，处理key续约问题
            new Timer().schedule(new KeyWatchdogTask(val,getKeyValue(val), KEY_EXPIRE_TIME_SECONDS, DELAY_WATCH_DOG), DELAY_WATCH_DOG);
        }
        return result;
    }

    private void tryLockWait(String val) {
        long currentTimeMillis = System.currentTimeMillis();
        Jedis jedis = JedisUtils.getJedis();

        if (tryLock(val)) {
            return;
        }

        // 订阅key,锁释放信号
        Future<String> subscribe = subscribe(val, jedis, new RedisKeyPubSub());
        try {
            // 会阻塞
            subscribe.get(TRY_LOCK_TIME_SECONDS, TimeUnit.SECONDS);
        } catch (InterruptedException | ExecutionException e) {
            e.printStackTrace();
            throw new ServiceException("获取锁异常");
        } catch (TimeoutException e) {
            throw new ServiceException("获取锁超时");
        }
        jedis = JedisUtils.getJedis();

        try {
            while (true) {
                // log.info(Thread.currentThread().getName() + "：尝试取锁.....");
                long timeMillis = System.currentTimeMillis();
                if (timeMillis - currentTimeMillis < TRY_LOCK_TIME_SECONDS * 1000) {
                    if (tryLock(val)) {
                        //log.info(Thread.currentThread().getName() + "：获取锁成功.....");
                        break;
                    }
                    try {
                        Thread.sleep(30);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                } else {
                    throw new ServiceException("获取锁超时");
                }
            }
        } finally {
            jedis.close();
        }
    }

    private Future<String> subscribe(String key, Jedis jedis, RedisKeyPubSub sub) {
        ExecutorService executorService = Executors.newFixedThreadPool(1);
        Future<String> submit = executorService.submit(() -> {
            AppExecutor.submit(() -> {
                jedis.subscribe(sub, key);
                jedis.close();
            });
            return sub.get();
        });
        executorService.shutdown();
        return submit;
    }

    @Override
    public void unlock(String variable) {
        Jedis jedis = JedisUtils.getJedis();
        try {
            // 使用脚本保证原子性操作,删除锁，发布信息
            String script =
                    "if redis.call('exists', KEYS[1]) == 0 then " +
                            "return -1;" +
                    " end; "+
                    "if redis.call('get', KEYS[1]) == ARGV[1] then "+
                           "redis.call('publish'," + "KEYS[1] ," + "ARGV[1]" + ");"+
                           "return redis.call('del', " + "KEYS[1]); " +
                    "else " +
                           "return 0; " +
                    "end";
            // log.info("释放锁,{},并且发布释放锁信号 publish", variable);
            String keyValue = getKeyValue(variable);
            Object result = jedis.eval(script, Arrays.asList(variable), Arrays.asList(keyValue));
            if (UNLOCK_RESULT_ERROR.equals(Integer.parseInt(result.toString()))) {
                throw new ServiceException("解锁失败，加锁与解锁者不同一个线程");
            }
        } finally {
            jedis.close();
        }
    }

    @Override
    public void sync(String val, Runnable runnable) {
        try {
            lock(val);
            runnable.run();
        } finally {
            this.unlock(val);
        }
    }

    private String getKeyValue(String val) {
        Thread thread = Thread.currentThread();
        return thread.getName() + ":" + thread.getId() + ":" + val;
    }

    private static void test() {
        log.info(Thread.currentThread().getName() + "：业务处理中.....");
        try {
            Thread.sleep(15*1000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        log.info(Thread.currentThread().getName() + "：业务处理完毕.....");
    }

    public static void main(String[] args) throws IOException {
        RedisVariableLock redisVariableLock = new RedisVariableLock();
        for (int i = 0; i < 1; i++) {
            int finalI = i;
            new Thread(() -> redisVariableLock.sync("key_12"+finalI, () -> test())).start();
        }
        System.in.read();
    }
}
